[FLINK-38336][state/forst] Avoid data copy during failover for ForSt statebackend #27042

AlexYinHan · 2025-09-25T12:54:24Z

What is the purpose of the change

This PR fixes FLINK-38336, by allowing ForSt statebackend to reuse the restored files in failover scenario.

Brief change log

Reorganize the local/remote path of ForSt into a ForStPathContainer
Determine whether we are in a failover scenario by comparing the DB path and the file paths stored in StateHandles
Use ReusableDataTransferStrategy if we are in a failover scenario

Verifying this change

This change added tests and can be verified as follows:

DataTransferStrategyTest#testBuildingStrategyAsExpected

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (no)

flinkbot · 2025-09-25T13:00:15Z

CI report:

9a2c99a Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

Zakelly · 2025-09-29T06:14:07Z

...s/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateBackend.java

-                                .getSharedStateDirectory();
+                FsCheckpointStorageAccess fsCheckpointStorageAccess =
+                        (FsCheckpointStorageAccess) env.getCheckpointStorageAccess();
+                remoteJobPath = fsCheckpointStorageAccess.getCheckpointsDirectory();


We can only share state from the shared directory, so what exactly the remoteJobPath means?

remoteJobPath is the parent directory that ends with 'JobID'.

For example:

remoteShareWithCheckpoint==True: If the ForSt base DIR is /checkpoints/jobid-xxx/shared/op_yyy__1_1__attempt_0, then remoteJobPath is /checkpoints/jobid-xxx.

...rst/src/main/java/org/apache/flink/state/forst/datatransfer/DataTransferStrategyBuilder.java

davidradl · 2025-10-03T11:41:10Z

.../flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStPathContainer.java

+            @Nullable Path remoteBasePath) {
+        this.localJobPath = localJobPath;
+        this.localBasePath = localBasePath;
+        this.localForStPath = localBasePath != null ? new Path(localBasePath, DB_DIR_STRING) : null;


we have localJobPath, localBasePath and localForStPath which is derived from localBasePath.

I was expecting localJobpath to be a subfolder of localBasePath, is this true? Is this the bean to validate that?

I see localBasePath can be null, in this case localForStPath is set to null, but localJobPath can have a value. What does this mean? The code indicates that it is possible to run without Forst but have a . Have I understood this correctly?

Yes, localJobpath is a subfolder of localBasePath.

basePath,jobPath are allowed to be null, just to stay consistent with the code before this PR. Currently the local paths can only be null in UT tests.

davidradl · 2025-10-03T11:43:31Z

.../flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStPathContainer.java

+
+import javax.annotation.Nullable;
+
+/** Container for ForSt paths. */


it would be worth describing each path here. The text talks of Forst paths , but there are only 2 of the 6 that mention Forst - localForStPath and remoteForStPath

Comments have been added to the code.

.../flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStPathContainer.java

davidradl · 2025-10-03T11:47:09Z

...nk-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java

+        if (pathContainer.getRemoteForStPath() != null
+                && pathContainer.getLocalForStPath() != null) {
+            if (cacheBasePath == null && pathContainer.getLocalBasePath() != null) {
+                cacheBasePath = new Path(pathContainer.getLocalBasePath().getPath(), "cache");


I assume thr log is incorrect, not should be now

The log is correct. It means the cache path is NOT specified in the configurations, so we set it to local base path.

...nk-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java

davidradl · 2025-10-03T11:50:18Z

...nk-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java

    public void forceClearRemoteDirectories() throws Exception {
-        if (remoteBasePath != null && remotePathNewlyCreated) {
-            clearDirectories(remoteBasePath);
+        if (pathContainer.getRemoteBasePath() != null && remotePathNewlyCreated) {


The forceClear does not seem to force anything, in fact it will only clear the directories in a specific case when remotePathNewlyCreated is set.

Yes, that is to fix FLINK-38433

davidradl · 2025-10-03T11:57:58Z

...ackend-forst/src/main/java/org/apache/flink/state/forst/sync/ForStSyncKeyedStateBackend.java

-                        optionsContainer.getLocalBasePath(),
-                        optionsContainer.getRemoteBasePath(),
-                        ex);
+                LOG.warn("Could not delete ForSt: {}.", optionsContainer.getPathContainer(), ex);


nit: it seems we lose information here. I read the log entry text before Could not delete ForSt local working directory {} to means it is only the local directory that could not be deleted and we put out the remote directory for information. It would be good to point to the actual folder that could not be deleted in the new log message - rather than list all the folders and not know which could not be deleted.

Actually, 'the actual folder that could not be deleted' should be included in the ex, so i think there is no info lost here.

...ends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStExtension.java

davidradl · 2025-10-03T12:00:30Z

...-forst/src/test/java/org/apache/flink/state/forst/datatransfer/DataTransferStrategyTest.java

+                RecoveryClaimMode.CLAIM,
+                CopyDataTransferStrategy.class);
+
+        testRestoreStrategyAsExpected(


parameterized test for all these permutations?

Yeah i've had a thought of that. However, these test cases have already been simple enough, and parameterizing them would not reduce the amount of code, but only make them harder to read.

davidradl · 2025-10-03T12:20:00Z

...nk-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackend.java

    @VisibleForTesting
    Path getLocalBasePath() {
-        return optionsContainer.getLocalBasePath();
+        return optionsContainer.getPathContainer().getLocalBasePath();


is optionsContainer.getPathContainer() ever null?

No. pathContainer is final and it should never be null

…statebackend

Zakelly

LGTM. Thanks for the fix

…statebackend (apache#27042) (cherry picked from commit 78f6e77)

AlexYinHan force-pushed the yh/dev_38336 branch 2 times, most recently from 9c4c31e to a8460bc Compare September 29, 2025 03:25

Zakelly reviewed Sep 29, 2025

View reviewed changes

...rst/src/main/java/org/apache/flink/state/forst/datatransfer/DataTransferStrategyBuilder.java Show resolved Hide resolved